package cn.ztuo.bitrade.Trade;

import cn.ztuo.bitrade.entity.*;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.math.BigDecimal;
import java.util.*;
@Slf4j
public class CoinTrader {
    private ContractCoin coin;
    private KafkaTemplate<String,String> kafkaTemplate;
    //交易币种的精度
    private int coinScale = 4;
    //基币的精度
    private int baseCoinScale = 4;
    private Logger logger = LoggerFactory.getLogger(CoinTrader.class);
    //买入限价订单链表，价格从高到低排列
    private TreeMap<BigDecimal, MergeOrder> buyLimitPriceQueue;
    //卖出限价订单链表，价格从低到高排列
    private TreeMap<BigDecimal, MergeOrder> sellLimitPriceQueue;
    //买入市价订单链表，按时间从小到大排序
    private LinkedList<ContractOrder> buyMarketQueue;
    //卖出市价订单链表，按时间从小到大排序
    private LinkedList<ContractOrder> sellMarketQueue;
    //卖盘盘口信息
    private TradePlate sellTradePlate;
    //买盘盘口信息
    private TradePlate buyTradePlate;
    //是否暂停交易
    private boolean tradingHalt = false;
    private boolean ready = false;

    public CoinTrader(ContractCoin coin){
        this.coin = coin;
    }

    /**
     * 初始化交易线程
     */
    public void initialize(){
        logger.info("init CoinTrader for symbol {}",coin.getSymbol());
        //买单队列价格降序排列
        buyLimitPriceQueue = new TreeMap<>(Comparator.reverseOrder());
        //卖单队列价格升序排列
        sellLimitPriceQueue = new TreeMap<>(Comparator.naturalOrder());
        buyMarketQueue = new LinkedList<>();
        sellMarketQueue = new LinkedList<>();
        sellTradePlate = new TradePlate(coin.getSymbol(), ContractOrderDirection.SELL);
        sellTradePlate.setCoinScale(coinScale);
        sellTradePlate.setBaseCoinScale(baseCoinScale);
        buyTradePlate = new TradePlate(coin.getSymbol(),ContractOrderDirection.BUY);
        buyTradePlate.setCoinScale(coinScale);
        buyTradePlate.setBaseCoinScale(baseCoinScale);
    }


    /**
     * 增加 机器人 到交易盘口 不做实际的处理;
     *
     * @param orders
     */
    public void addRobotLimitPriceOrder(List<ContractOrder> orders, ContractOrderDirection direction,ContractCoin coin) {

        logger.info("addRobotLimitPriceOrder,orders = {}", orders);
        if (direction == ContractOrderDirection.BUY) {
            orders.stream().forEach(order->{
                buyTradePlate.add(order,coin);
            });
            if (ready) {
                sendTradePlateMessage(buyTradePlate);
            }
        }
        if (direction == ContractOrderDirection.SELL) {
            orders.stream().forEach(order->{
                sellTradePlate.add(order,coin);
            });
            if (ready) {
                sendTradePlateMessage(sellTradePlate);
            }
        }
    }

    /**
     * 取消盘口 机器人委托订单 无实际订单的处理;
     *
     * @param orders
     * @return
     */
    public Boolean cancelRobotOrder(List<ContractOrder> orders, ContractOrderDirection direction,ContractCoin coin) {
        logger.info("orderCanceled,orders={}", orders);
        if (direction == ContractOrderDirection.BUY) {
            orders.stream().forEach(order->{
                buyTradePlate.remove(order,coin);
            });
            sendTradePlateMessage(buyTradePlate);
        }
        if (direction == ContractOrderDirection.SELL) {
            orders.stream().forEach(order->{
                sellTradePlate.remove(order,coin);
            });
            sendTradePlateMessage(sellTradePlate);
        }
        return true;
    }
    
    
    /**
     * 增加限价订单到队列，买入单按从价格高到低排，卖出单按价格从低到高排
     * @param contractOrder
     */
    public void addLimitPriceOrder(ContractOrder contractOrder){
        if(contractOrder.getType() == ContractOrderType.MARKET_PRICE){
            return ;
        }
        logger.info("addLimitPriceOrder,orderId = {}", contractOrder.getOrderId());
        TreeMap<BigDecimal, MergeOrder> list;
        if(contractOrder.getDirection() == ContractOrderDirection.BUY){
            list = buyLimitPriceQueue;
            //将订单信息压入买入盘口 (根据订单价格进行排序)
            buyTradePlate.add(contractOrder,coin);
            if(ready) {
                sendTradePlateMessage(buyTradePlate);
            }
        }
        else {
            list = sellLimitPriceQueue;
            //将订单信息压入卖出盘口 (根据订单价格进行排序)
            sellTradePlate.add(contractOrder,coin);
            if(ready) {
                sendTradePlateMessage(sellTradePlate);
            }
        }
        //合并限价订单 并 加入订单到队列中
        synchronized (list) {
            MergeOrder mergeOrder = list.get(contractOrder.getPrice());
            //如果没有该价格 则新增 一个价格帧
            if(mergeOrder == null){
                mergeOrder = new MergeOrder();
                mergeOrder.add(contractOrder);
                list.put(contractOrder.getPrice(), mergeOrder);
            }
            //将该订单加入 此价格栈
            else {
                mergeOrder.add(contractOrder);
            }
        }
    }
    //市场价订单 不压入价格栈 也不加入 盘口列表
    public void addMarketPriceOrder(ContractOrder contractOrder){
        if(contractOrder.getType() != ContractOrderType.MARKET_PRICE){
            return ;
        }
        logger.info("addMarketPriceOrder,orderId = {}", contractOrder.getOrderId());
        LinkedList<ContractOrder> list = contractOrder.getDirection() == ContractOrderDirection.BUY ? buyMarketQueue : sellMarketQueue;
        synchronized (list) {
            list.addLast(contractOrder);
        }
    }

    public void trade(List<ContractOrder> orders){
        if(tradingHalt) {
            return ;
        }
        for(ContractOrder order:orders){
            //主动交易输入的订单，交易不完成的会输入到队列   jhh
            trade(order);
        }
    }


    /**
     * 主动交易输入的订单，交易不完成的会输入到队列
     * 该项目最核心的方法  ------  交易
     * @param contractOrder
     */

    public void trade(ContractOrder contractOrder){
        if(tradingHalt){
            return ;
        }
        logger.info("trade order={}",contractOrder);
        if(!coin.getSymbol().equalsIgnoreCase(contractOrder.getSymbol())){
            logger.info("unsupported symbol,coin={},base={}", contractOrder.getCoinSymbol(), contractOrder.getBaseSymbol());
            return ;
        }
        if(contractOrder.getAmount().compareTo(BigDecimal.ZERO) <=0 || contractOrder.getAmount().subtract(contractOrder.getTradedAmount()).compareTo(BigDecimal.ZERO)<=0){
            return ;
        }

        TreeMap<BigDecimal, MergeOrder> limitPriceOrderList;
        LinkedList<ContractOrder> marketPriceOrderList;
        if(contractOrder.getDirection() == ContractOrderDirection.BUY){
            limitPriceOrderList = sellLimitPriceQueue;
            marketPriceOrderList = sellMarketQueue;
        }else{
            limitPriceOrderList = buyLimitPriceQueue;
            marketPriceOrderList = buyMarketQueue;
        }
        //挂单类型== 市价
        if(contractOrder.getType() == ContractOrderType.MARKET_PRICE){
            //市价委托单与限价对手单列表交易  jhh
                matchMarketPriceWithLPList(limitPriceOrderList, contractOrder);
        }else if(contractOrder.getType() != ContractOrderType.MARKET_PRICE){
            //限价单价格必须大于0
            if(contractOrder.getPrice().compareTo(BigDecimal.ZERO) <= 0){
                return ;
            }
            //限价委托单与限价队列匹配  jhh
            matchLimitPriceWithLPList(limitPriceOrderList, contractOrder,false);
            //买入或卖出量，对于市价买入单表 大于 成交量  jhh
            if(contractOrder.getAmount().compareTo(contractOrder.getTradedAmount()) > 0) {
                //限价委托单与市价队列匹配  jhh
                matchLimitPriceWithMPList(marketPriceOrderList, contractOrder);
            }
        }
    }

    /**
     * 限价委托单与限价队列匹配
     * @param lpList 限价对手单队列
     * @param focusedOrder 交易订单
     */
    public void matchLimitPriceWithLPList(TreeMap<BigDecimal, MergeOrder> lpList, ContractOrder focusedOrder, boolean canEnterList){
        List<ContractTrade> contractTrades = new ArrayList<>();
        //已完成的订单集合
        List<ContractOrder> completedOrders = new ArrayList<>();
        synchronized (lpList) {
            Iterator<Map.Entry<BigDecimal, MergeOrder>> mergeOrderIterator = lpList.entrySet().iterator();
            boolean exitLoop = false;
            while (!exitLoop && mergeOrderIterator.hasNext()) {
                Map.Entry<BigDecimal, MergeOrder> entry = mergeOrderIterator.next();
                MergeOrder mergeOrder = entry.getValue();
                Iterator<ContractOrder> orderIterator = mergeOrder.iterator();
                //买入单需要匹配的价格不大于委托价，否则退出
                if (focusedOrder.getDirection() == ContractOrderDirection.BUY && mergeOrder.getPrice().compareTo(focusedOrder.getPrice()) > 0) {
                    break;
                }
                //卖出单需要匹配的价格不小于委托价，否则退出
                if (focusedOrder.getDirection() == ContractOrderDirection.SELL && mergeOrder.getPrice().compareTo(focusedOrder.getPrice()) < 0) {
                    break;
                }
                while (orderIterator.hasNext()) {
                    ContractOrder matchOrder = orderIterator.next();
                    //memberId=1 机器人 只能 和自己进行 匹配 否则 continue
                    if((focusedOrder.getMemberId()==1&&matchOrder.getMemberId()!=1)||(focusedOrder.getMemberId()!=1&&matchOrder.getMemberId()==1)){
                        continue;
                    }
                    //处理两个匹配的委托订单   jhh
                    ContractTrade trade = processMatch(focusedOrder, matchOrder);
                    contractTrades.add(trade);
                    //判断匹配单是否完成
                    if (matchOrder.isCompleted()) {
                        //当前匹配的订单完成交易，删除该订单
                        orderIterator.remove();
                        completedOrders.add(matchOrder);
                    }
                    //判断交易单是否完成
                    if (focusedOrder.isCompleted()) {
                        //交易完成
                        completedOrders.add(focusedOrder);
                        //退出循环
                        exitLoop = true;
                        break;
                    }
                }
                if(mergeOrder.size() == 0){
                    mergeOrderIterator.remove();
                }
            }
        }
        //如果还没有交易完，订单压入列表中
        if (focusedOrder.getTradedAmount().compareTo(focusedOrder.getAmount()) < 0 && canEnterList) {
//        if (focusedOrder.getTradedAmount().compareTo(focusedOrder.getAmount()) < 0 && canEnterList) {
            addLimitPriceOrder(focusedOrder);
        }
        //处理成交明细
        handlecontractTrade(contractTrades);

        if(completedOrders.size() > 0){
            orderCompleted(completedOrders);
            TradePlate plate = focusedOrder.getDirection() == ContractOrderDirection.BUY ? sellTradePlate : buyTradePlate;
            sendTradePlateMessage(plate);
        }
    }

    /**
     * 限价委托单与市价队列匹配
     * @param mpList 市价对手单队列
     * @param focusedOrder 交易订单
     */
    public void matchLimitPriceWithMPList(LinkedList<ContractOrder> mpList,ContractOrder focusedOrder){
        List<ContractTrade> contractTrades = new ArrayList<>();
        List<ContractOrder> completedOrders = new ArrayList<>();
        //线程处理订单 jhh
        synchronized (mpList) {
            //.iterator() 类似for循环  jhh
            Iterator<ContractOrder> iterator = mpList.iterator();
            // 使用hasNext()检查序列中是否还有元素。jhh
            while (iterator.hasNext()) {
                //.next() 去下一个数据使用next()获得序列中的下一个元素。jhh
                ContractOrder matchOrder = iterator.next();
                //memberId=1 机器人 只能 和自己进行 匹配 否则 continue
                if((focusedOrder.getMemberId()==1&&matchOrder.getMemberId()!=1)||(focusedOrder.getMemberId()!=1&&matchOrder.getMemberId()==1)){
                    continue;
                }
                ContractTrade trade = processMatch(focusedOrder, matchOrder);
                if(trade != null){
                    contractTrades.add(trade);
                }
                //判断匹配单是否完成，市价单amount为成交量
                if(matchOrder.isCompleted()){
                    //使用remove()将迭代器新返回的元素删除。 jhh
                    iterator.remove();
                    completedOrders.add(matchOrder);
                }
                //判断吃单是否完成，判断成交量是否完成
                if (focusedOrder.isCompleted()) {
                    //交易完成
                    completedOrders.add(focusedOrder);
                    //退出循环
                    break;
                }
            }
        }
        //如果还没有交易完，订单压入列表中
        if (focusedOrder.getTradedAmount().compareTo(focusedOrder.getAmount()) < 0) {
            addLimitPriceOrder(focusedOrder);
        }
        //每个订单的匹配批量推送
        handlecontractTrade(contractTrades);
        orderCompleted(completedOrders);
    }


    /**
     * 市价委托单与限价对手单列表交易
     * @param lpList  限价对手单列表
     * @param focusedOrder 待交易订单
     */
    public void matchMarketPriceWithLPList(TreeMap<BigDecimal, MergeOrder> lpList, ContractOrder focusedOrder){
        List<ContractTrade> contractTrades = new ArrayList<>();
        //已完成的订单集合 jjhh
        List<ContractOrder> completedOrders = new ArrayList<>();
        //线程处理 jjhh
        synchronized (lpList) {
            // .iterator() 获取元素 jjhh
            Iterator<Map.Entry<BigDecimal, MergeOrder>> mergeOrderIterator = lpList.entrySet().iterator();
            boolean exitLoop = false;
            //.hasNext() 检查 是否还有元素 jhh
            while (!exitLoop && mergeOrderIterator.hasNext()) {
                // .next() 获取下一个元素 jhh
                Map.Entry<BigDecimal, MergeOrder> entry = mergeOrderIterator.next();
                MergeOrder mergeOrder = entry.getValue();
                Iterator<ContractOrder> orderIterator = mergeOrder.iterator();
                while (orderIterator.hasNext()) {
                    ContractOrder matchOrder = orderIterator.next();

                    //memberId=1 机器人 只能 和自己进行 匹配 否则 continue
                    if((focusedOrder.getMemberId()==1&&matchOrder.getMemberId()!=1)||(focusedOrder.getMemberId()!=1&&matchOrder.getMemberId()==1)){
                        continue;
                    }
                    //处理两个匹配的委托订单  jhh
                    ContractTrade trade = processMatch(focusedOrder, matchOrder);
                    //撮合完成跳过  没有交易完  添加到集合 jhh
                    if (trade != null) {
                        contractTrades.add(trade);
                    }
                    //判断匹配单是否完成
                    if (matchOrder.isCompleted()) {
                        //当前匹配的订单完成交易，删除该订单
                        orderIterator.remove();
                        completedOrders.add(matchOrder);
                    }
                    //判断焦点订单是否完成
                    if (focusedOrder.isCompleted()) {
                        completedOrders.add(focusedOrder);
                        //退出循环
                        exitLoop = true;
                        break;
                    }
                }
                if(mergeOrder.size() == 0){
                    mergeOrderIterator.remove();
                }
            }
        }
        //如果还没有交易完，订单压入列表中,市价买单按成交量算
        if (focusedOrder.getDirection() == ContractOrderDirection.SELL&&focusedOrder.getTradedAmount().compareTo(focusedOrder.getAmount()) < 0
                || focusedOrder.getDirection() == ContractOrderDirection.BUY&& focusedOrder.getTurnover().compareTo(focusedOrder.getAmount()) < 0) {
            addMarketPriceOrder(focusedOrder);
        }
        //每个订单的匹配批量推送
        handlecontractTrade(contractTrades);
        //完成的订单数量大于0  推送pc展示 jhh
        if(completedOrders.size() > 0){
            //订单完成，执行消息通知  jhh
            orderCompleted(completedOrders);
            TradePlate plate = focusedOrder.getDirection() == ContractOrderDirection.BUY ? sellTradePlate : buyTradePlate;
            //推送盘口展示  jhh
            sendTradePlateMessage(plate);
        }
    }

    /**
     * 计算委托单剩余可成交的数量
     * @param order 委托单
     * @param dealPrice 成交价
     * @return
     */
    private BigDecimal calculateTradedAmount(ContractOrder order, BigDecimal dealPrice){
//        if(order.getDirection() == ContractOrderDirection.BUY && order.getType() == ContractOrderType.MARKET_PRICE){
//            //剩余成交量
//            BigDecimal leftTurnover = order.getAmount().subtract(order.getTurnover());
//            return leftTurnover.divide(dealPrice,coinScale,BigDecimal.ROUND_DOWN);
//        }
//        else{
            return  order.getAmount().subtract(order.getTradedAmount());
//        }
    }

    /**
     * 调整市价单剩余成交额，当剩余成交额不足时设置订单完成
     * @param order
     * @param dealPrice
     * @return
     */
    private BigDecimal adjustMarketOrderTurnoverAmount(ContractOrder order, BigDecimal dealPrice){
        if(order.getDirection() ==ContractOrderDirection.BUY && order.getType() == ContractOrderType.MARKET_PRICE){
            BigDecimal leftTurnoverAmount = order.getAmount().subtract(order.getTradedAmount());
//            if(leftTurnoverAmount.divide(dealPrice,coinScale,BigDecimal.ROUND_DOWN)
//                .compareTo(BigDecimal.ZERO)==0){
//                order.setTurnover(order.getAmount());
                return leftTurnoverAmount;
//            }
        }
        return BigDecimal.ZERO;
    }

    /**
     * 处理两个匹配的委托订单
     * @param focusedOrder 焦点单
     * @param matchOrder 匹配单
     * @return
     */
    private ContractTrade processMatch(ContractOrder focusedOrder, ContractOrder matchOrder){
        //需要交易的数量，成交量,成交价，可用数量
        BigDecimal needAmount,dealPrice,availAmount;
        //如果匹配单是限价单，则以其价格为成交价
        if(matchOrder.getType() != ContractOrderType.MARKET_PRICE){
            dealPrice = matchOrder.getPrice();
        } else {
            dealPrice = focusedOrder.getPrice();
        }
        //成交价必须大于0
        if(dealPrice.compareTo(BigDecimal.ZERO) <= 0){
            return null;
        }
        //计算委托单剩余可成交的数量  jhh
        needAmount = calculateTradedAmount(focusedOrder,dealPrice);
        //计算委托单剩余可用数量  jhh
        availAmount = calculateTradedAmount(matchOrder,dealPrice);
        //计算成交量
        BigDecimal tradedAmount = (availAmount.compareTo(needAmount) >= 0 ? needAmount : availAmount);
        //如果成交额为0说明剩余额度无法成交，退出
        if(tradedAmount.compareTo(BigDecimal.ZERO) == 0){
            return null;
        }

        //计算成交额,成交额要保留足够精度
        BigDecimal turnover = tradedAmount.multiply(dealPrice);

        //计算成交额所对应的保证金
        BigDecimal focusedTradeBond = focusedOrder.getStart().multiply(tradedAmount.divide(needAmount,8,BigDecimal.ROUND_DOWN));
        BigDecimal matchTradeBond = matchOrder.getStart().multiply(tradedAmount.divide(availAmount,8,BigDecimal.ROUND_DOWN));
        log.info(focusedOrder.getPrice().toString()+"-------focusedTradeBond :"+focusedTradeBond.toString());
        log.info(matchOrder.getPrice().toString()+"-------matchTradeBond :"+matchTradeBond.toString());
        log.info("matchOrder.getStart()-------matchTradeBond :"+matchOrder.getStart().toString());

        if(matchOrder.getAmount().compareTo(matchOrder.getTradedAmount().add(tradedAmount))==0){
            matchTradeBond=matchOrder.getStart();
            matchOrder.setStart(BigDecimal.ZERO);
        }else{
            matchOrder.setStart(matchOrder.getStart().subtract(matchTradeBond));
        }
        matchOrder.setTradedAmount(matchOrder.getTradedAmount().add(tradedAmount));
        log.info("matchOrder.getTurnover()"+matchOrder.getTurnover()+"-------------------------------------------------");
        matchOrder.setTurnover(matchOrder.getTurnover().add(turnover));

        if(focusedOrder.getAmount().compareTo(focusedOrder.getTradedAmount().add(tradedAmount))==0){
            focusedTradeBond=focusedOrder.getStart();
            focusedOrder.setStart(BigDecimal.ZERO);
        }else{
            focusedOrder.setStart(focusedOrder.getStart().subtract(focusedTradeBond));
        }
        focusedOrder.setTradedAmount(focusedOrder.getTradedAmount().add(tradedAmount));
        focusedOrder.setTurnover(focusedOrder.getTurnover().add(turnover));

        //创建成交记录
        ContractTrade contractTrade = new ContractTrade();
        contractTrade.setSymbol(coin.getSymbol());
        contractTrade.setAmount(tradedAmount);
        contractTrade.setNowp(tradedAmount.divide(coin.getMinVolume(),0,BigDecimal.ROUND_DOWN));
        contractTrade.setDirection(focusedOrder.getDirection());
        contractTrade.setPrice(dealPrice);
        contractTrade.setBuyTurnover(turnover);
        contractTrade.setSellTurnover(turnover);
        //校正市价单剩余成交额
//        if(ContractOrderType.MARKET_PRICE == focusedOrder.getType() && focusedOrder.getDirection() == ContractOrderDirection.BUY){
//            //调整市价单剩余成交额，当剩余成交额不足时设置订单完成  jhh
//            BigDecimal adjustTurnover = adjustMarketOrderTurnoverAmount(focusedOrder,dealPrice);
//            contractTrade.setBuyTurnover(turnover.add(adjustTurnover));
//        }
//        else if(ContractOrderType.MARKET_PRICE == matchOrder.getType() && matchOrder.getDirection() == ContractOrderDirection.BUY){
//            BigDecimal adjustTurnover = adjustMarketOrderTurnoverAmount(matchOrder,dealPrice);
//            contractTrade.setBuyTurnover(turnover.add(adjustTurnover));
//        }

        if (focusedOrder.getDirection() == ContractOrderDirection.BUY) {
            contractTrade.setBuyOrderId(focusedOrder.getOrderId());
            contractTrade.setSellOrderId(matchOrder.getOrderId());

            contractTrade.setBuyBond(focusedTradeBond);
            contractTrade.setSellBond(matchTradeBond);
        } else {
            contractTrade.setBuyOrderId(matchOrder.getOrderId());
            contractTrade.setSellOrderId(focusedOrder.getOrderId());

            contractTrade.setBuyBond(matchTradeBond);
            contractTrade.setSellBond(focusedTradeBond);
        }

        contractTrade.setTime(Calendar.getInstance().getTimeInMillis());
        if(matchOrder.getType() != ContractOrderType.MARKET_PRICE){
            if(matchOrder.getDirection() == ContractOrderDirection.BUY){
                buyTradePlate.remove(matchOrder,tradedAmount,coin);
            }
            else{
                sellTradePlate.remove(matchOrder,tradedAmount,coin);
            }
        }


        return  contractTrade;
    }

    /**
     * 撮合交易
     * @param trades
     */
    public void handlecontractTrade(List<ContractTrade> trades){
        logger.info("handlecontractTrade:{}", trades);
        if(trades.size() > 0) {
            int maxSize = 1000;
            //发送消息，key为交易对符号
            if(trades.size() > maxSize) {
                int size = trades.size();
                for(int index = 0;index < size;index += maxSize){
                    int length = (size - index) > maxSize ? maxSize : size - index;
                    List<ContractTrade> subTrades = trades.subList(index,index + length);
                    kafkaTemplate.send("contract-trade", coin.getSymbol(), JSON.toJSONString(subTrades));
                    logger.info("拆分发送订单成交明细消息={}",coin.getSymbol());
                }
            }
            else {
                kafkaTemplate.send("contract-trade", coin.getSymbol(), JSON.toJSONString(trades));
                logger.info("发送订单成交明细消息={}",coin.getSymbol());
            }
        }
    }

    /**
     * 订单完成，执行消息通知,订单数超1000个要拆分发送
     * @param orders
     */
    public  void orderCompleted(List<ContractOrder> orders){
        logger.info("orderCompleted ,order={}",orders);
        if(orders.size() > 0) {
            int maxSize = 1000;
            if(orders.size() > maxSize){
                int size = orders.size();
                for(int index = 0;index < size;index += maxSize){
                    int length = (size - index) > maxSize ? maxSize : size - index;
                    List<ContractOrder> subOrders = orders.subList(index,index + length);
                    kafkaTemplate.send("contract-order-completed", coin.getSymbol(), JSON.toJSONString(subOrders));
                    logger.info("拆分发送订单完成消息={}",coin.getSymbol());
                }
            } else {
                kafkaTemplate.send("contract-order-completed", coin.getSymbol(), JSON.toJSONString(orders));
                logger.info("发送订单完成消息={}",coin.getSymbol());
            }
        }
    }

    /**
     * 发送盘口变化消息
     * @param plate
     */
    public void sendTradePlateMessage(TradePlate plate){
        kafkaTemplate.send("contract-trade-plate", coin.getSymbol(), JSON.toJSONString(plate));
        logger.info("发送盘口变化消息={}",coin.getSymbol());
    }

    /**
     * 取消委托订单
     * @param contractOrder
     * @return
     */
    public ContractOrder cancelOrder(ContractOrder contractOrder){
        logger.info("orderCanceled,orderId={}", contractOrder.getOrderId());
        if(contractOrder.getType() == ContractOrderType.MARKET_PRICE){
            //处理市价单
            Iterator<ContractOrder> orderIterator;
            List<ContractOrder> list = null;
            if(contractOrder.getDirection() == ContractOrderDirection.BUY){
                list = this.buyMarketQueue;
            }
            else{
                list = this.sellMarketQueue;
            }
            synchronized (list) {
                orderIterator = list.iterator();
                while ((orderIterator.hasNext())) {
                    ContractOrder order = orderIterator.next();
                    if (order.getOrderId().equalsIgnoreCase(contractOrder.getOrderId())) {
                        orderIterator.remove();
                        onRemoveOrder(order);
                        return order;
                    }
                }
            }
        }
        else {
            //处理限价单
            TreeMap<BigDecimal, MergeOrder> list = null;
            Iterator<MergeOrder> mergeOrderIterator;
            if(contractOrder.getDirection() == ContractOrderDirection.BUY){
                list = this.buyLimitPriceQueue;
            }
            else{
                list = this.sellLimitPriceQueue;
            }
            synchronized (list) {
                MergeOrder mergeOrder = list.get(contractOrder.getPrice());
                if(mergeOrder==null){
                    return null;
                }
                Iterator<ContractOrder> orderIterator = mergeOrder.iterator();
                if (orderIterator != null) {
                    while (orderIterator.hasNext()) {
                        ContractOrder order = orderIterator.next();
                        if (order.getOrderId().equalsIgnoreCase(contractOrder.getOrderId())) {
                            orderIterator.remove();
                            if (mergeOrder.size() == 0) {
                                list.remove(contractOrder.getPrice());
                            }
                            onRemoveOrder(order);
                            return order;
                        }
                    }
                }
            }
        }
        return  null;
    }

    public void onRemoveOrder(ContractOrder order){
        if (order.getType() != ContractOrderType.MARKET_PRICE) {
            if (order.getDirection() == ContractOrderDirection.BUY) {
                buyTradePlate.remove(order,coin);
                sendTradePlateMessage(buyTradePlate);
            } else {
                sellTradePlate.remove(order,coin);
                sendTradePlateMessage(sellTradePlate);
            }
        }
    }



    public TradePlate getTradePlate(ContractOrderDirection direction){
        if(direction == ContractOrderDirection.BUY){
            return buyTradePlate;
        }
        else{
            return sellTradePlate;
        }
    }



    /**
     * 查询交易器里的订单
     * @param orderId
     * @param type
     * @param direction
     * @return
     */
    public ContractOrder findOrder(String orderId,ContractOrderType type,ContractOrderDirection direction){
        if(type == ContractOrderType.MARKET_PRICE){
            LinkedList<ContractOrder> list;
            if(direction == ContractOrderDirection.BUY){
                list = this.buyMarketQueue;
            }
            else{
                list = this.sellMarketQueue;
            }
            synchronized (list) {
                Iterator<ContractOrder> orderIterator = list.iterator();
                while ((orderIterator.hasNext())) {
                    ContractOrder order = orderIterator.next();
                    if (order.getOrderId().equalsIgnoreCase(orderId)) {
                        return order;
                    }
                }
            }
        }
        else {
            TreeMap<BigDecimal, MergeOrder> list;
            if(direction == ContractOrderDirection.BUY){
                list = this.buyLimitPriceQueue;
            }
            else{
                list = this.sellLimitPriceQueue;
            }
            synchronized (list) {
                Iterator<Map.Entry<BigDecimal, MergeOrder>> mergeOrderIterator = list.entrySet().iterator();
                while (mergeOrderIterator.hasNext()) {
                    Map.Entry<BigDecimal, MergeOrder> entry = mergeOrderIterator.next();
                    MergeOrder mergeOrder = entry.getValue();
                    Iterator<ContractOrder> orderIterator = mergeOrder.iterator();
                    while ((orderIterator.hasNext())) {
                        ContractOrder order = orderIterator.next();
                        if (order.getOrderId().equalsIgnoreCase(orderId)) {
                            return order;
                        }
                    }
                }
            }
        }
        return null;
    }

    public TreeMap<BigDecimal, MergeOrder> getBuyLimitPriceQueue() {
        return buyLimitPriceQueue;
    }

    public LinkedList<ContractOrder> getBuyMarketQueue() {
        return buyMarketQueue;
    }

    public TreeMap<BigDecimal, MergeOrder> getSellLimitPriceQueue() {
        return sellLimitPriceQueue;
    }

    public LinkedList<ContractOrder> getSellMarketQueue() {
        return sellMarketQueue;
    }

    public void setKafkaTemplate(KafkaTemplate<String,String> template){
        this.kafkaTemplate = template;
    }
    public void setCoinScale(int scale){
        this.coinScale = scale;
    }

    public void setBaseCoinScale(int scale){
        this.baseCoinScale = scale;
    }

    public boolean isTradingHalt(){
        return this.tradingHalt;
    }

    /**
     * 暂停交易,不接收新的订单
     */
    public void haltTrading(){
        this.tradingHalt = true;
    }

    /**
     * 恢复交易
     */
    public void resumeTrading(){
        this.tradingHalt = false;
    }

    public void stopTrading(){
        //TODO:停止交易，取消当前所有订单
    }

    public boolean getReady(){
        return this.ready;
    }

    public void setReady(boolean ready){
        this.ready = ready;
    }

    public int getLimitPriceOrderCount(ContractOrderDirection direction){
        int count = 0;
        TreeMap<BigDecimal, MergeOrder> queue = direction == ContractOrderDirection.BUY ? buyLimitPriceQueue : sellLimitPriceQueue;
        Iterator<Map.Entry<BigDecimal, MergeOrder>> mergeOrderIterator = queue.entrySet().iterator();
        while (mergeOrderIterator.hasNext()) {
            Map.Entry<BigDecimal, MergeOrder> entry = mergeOrderIterator.next();
            MergeOrder mergeOrder = entry.getValue();
            count += mergeOrder.size();
        }
        return count;
    }
}
